/* contrib/xlogminer/xlogminer--1.0.sql */
CREATE OR REPLACE FUNCTION pg_minerwal(starttime text, endtime text, startxid int, endxid int, startlsn pg_lsn, endlsn pg_lsn, tempresult bool, reloid  Oid, use_history_image bool = false)
RETURNS text AS
'MODULE_PATHNAME','pg_minerXlog'
LANGUAGE C CALLED ON NULL INPUT;

CREATE OR REPLACE FUNCTION walminer_temp_table_check()
RETURNS void AS
$BODY$
DECLARE
	rd "varchar";
	checksql "varchar";
	temptablename "varchar";
	tp "varchar";
BEGIN
	temptablename := 'walminer_contents';
	tp :='t';
	SELECT * into rd FROM pg_catalog.pg_class WHERE relname = 'walminer_contents' AND relpersistence = 't' AND pg_table_is_visible(oid);
	IF FOUND THEN
		TRUNCATE TABLE walminer_contents;
	ELSE
		CREATE temp TABLE walminer_contents(xid bigint,virtualxid int,timestampTz timestampTz,record_database text, record_user text,record_tablespace text,record_schema text, op_type text,op_text text,op_undo text,lsn pg_lsn,commit_end_lsn pg_lsn);
	END IF;
END;
$BODY$
LANGUAGE 'plpgsql' VOLATILE; 

CREATE OR REPLACE FUNCTION walminer_start(startlsn pg_lsn, endlsn pg_lsn, tempresult bool DEFAULT 'false', use_history_image bool = false)
RETURNS text AS 
$BODY$
	select walminer_temp_table_check();
	select pg_minerwal('NULL', 'NULL', 0, 0, $1,$2,$3,0,$4);
$BODY$
LANGUAGE 'sql';

CREATE OR REPLACE FUNCTION walminer_start(starttime text, endtime text, startxid int, endxid int, tempresult bool DEFAULT 'false', use_history_image bool = false)
RETURNS text AS 
$BODY$
	select walminer_temp_table_check();
	select pg_minerwal($1,$2,$3,$4,NULL,NULL,$5,0,$6);
$BODY$
LANGUAGE 'sql';

CREATE OR REPLACE FUNCTION walminer_start_simpletable(reloid Oid,tempresult bool DEFAULT 'false', use_history_image bool = false)
RETURNS text AS 
$BODY$
	select walminer_temp_table_check();
	select pg_minerwal('NULL', 'NULL', 0, 0, NULL, NULL, $2, $1, $3);
$BODY$
LANGUAGE 'sql';

CREATE OR REPLACE FUNCTION walminer_build_dictionary(in path text)
RETURNS text AS
'MODULE_PATHNAME','xlogminer_build_dictionary'
LANGUAGE C CALLED ON NULL INPUT;

CREATE OR REPLACE FUNCTION walminer_load_dictionary(in path text)
RETURNS text AS
'MODULE_PATHNAME','xlogminer_load_dictionary'
LANGUAGE C CALLED ON NULL INPUT;

CREATE OR REPLACE FUNCTION walminer_stop()
RETURNS text AS
'MODULE_PATHNAME','xlogminer_stop'
LANGUAGE C VOLATILE STRICT;

CREATE OR REPLACE FUNCTION walminer_wal_add(in path text,startlsn pg_lsn = NULL, endlsn pg_lsn = NULL, startlsn_scan_seg_advance int = -1)
RETURNS text AS
'MODULE_PATHNAME','xlogminer_xlogfile_add'
LANGUAGE C CALLED ON NULL INPUT;

CREATE OR REPLACE FUNCTION walminer_wal_remove(in path text)
RETURNS text AS
'MODULE_PATHNAME','xlogminer_xlogfile_remove'
LANGUAGE C CALLED ON NULL INPUT;

CREATE OR REPLACE FUNCTION walminer_wal_list()
RETURNS setof record  AS
'MODULE_PATHNAME','xlogminer_xlogfile_list'
LANGUAGE C VOLATILE STRICT;


-- 设置startlsn_scan_seg_advance为-1禁用WAL文件过滤
CREATE OR REPLACE FUNCTION wal2sql(wal_dir text, startlsn pg_lsn, endlsn pg_lsn = NULL, startlsn_scan_seg_advance int = 2, use_history_image bool = false, tempresult bool = false)
RETURNS TABLE(xid bigint,virtualxid int,timestampTz timestampTz,record_database text, record_user text,record_tablespace text,record_schema text, op_type text,op_text text,op_undo text,lsn pg_lsn,commit_end_lsn pg_lsn) AS 
$BODY$
DECLARE
	wal_dir_tmp  text;
BEGIN
	perform walminer_stop();
	
	if wal_dir is NULL then 
		select setting||'/pg_wal' into wal_dir_tmp from pg_settings where name='data_directory';
	else
		wal_dir_tmp := wal_dir;
	end if;
	
	perform walminer_wal_add(wal_dir_tmp,startlsn,endlsn,startlsn_scan_seg_advance);
	
	-- 当前解析WAL时，先快速扫描事务记录，满足条件时在回溯到上一个事务的开始位置正式扫描收集SQL，事务并发执行时，容易遗漏SQL
	-- 通过设置合理的startlsn_scan_seg_advance除提升解析性能，也可以在一定程度上回避这个问题，除非一个事务跨越的WAL文件数大于startlsn_scan_seg_advance。
	if startlsn is not null and startlsn_scan_seg_advance >=0 then
		perform walminer_start(NULL, endlsn, tempresult, use_history_image);
	-- 并发场景下，解析出的SQL顺序会乱，如果改内部逻辑涉及东西太多，所以临时用order by回避
		return query select * from walminer_contents c where c.commit_end_lsn > startlsn order by commit_end_lsn,lsn;
	else
	-- 如果走这个分支，上面提到的bug是无法避免的
		perform walminer_start(startlsn, endlsn, tempresult, use_history_image);
		return query select * from walminer_contents order by commit_end_lsn,lsn;
	end if;
	
	perform walminer_stop();
END;
$BODY$
LANGUAGE 'plpgsql' VOLATILE;


-- 基于时间范围解析WAL
CREATE OR REPLACE FUNCTION wal2sql_bytime(wal_dir text, starttime timestampTz, endtime timestampTz = NULL, use_history_image bool = false, tempresult bool = false)
RETURNS TABLE(xid bigint,virtualxid int,timestampTz timestampTz,record_database text, record_user text,record_tablespace text,record_schema text, op_type text,op_text text,op_undo text,lsn pg_lsn,commit_end_lsn pg_lsn) AS 
$BODY$
DECLARE
	wal_dir_tmp  text;
	starttime_text  text;
	endtime_text  text;
BEGIN
	perform walminer_stop();
	
	if wal_dir is NULL then 
		select setting||'/pg_wal' into wal_dir_tmp from pg_settings where name='data_directory';
	else
		wal_dir_tmp := wal_dir;
	end if;
	
	perform walminer_wal_add(wal_dir_tmp);

	if starttime is NULL then
		starttime_text := 'NULL';
	else
		select starttime::text into starttime_text;
	end if;

	if endtime is NULL then
		endtime_text := 'NULL';
	else
		select endtime::text into endtime_text;
	end if;
	

	perform walminer_start(starttime_text, endtime_text, 0, 0, tempresult, use_history_image);
	return query select * from walminer_contents order by commit_end_lsn,lsn;
	
	perform walminer_stop();
END;
$BODY$
LANGUAGE 'plpgsql' VOLATILE;


-- 基于xid范围解析WAL
CREATE OR REPLACE FUNCTION wal2sql_byxid(wal_dir text, startxid int, endxid int = 0, use_history_image bool = false, tempresult bool = false)
RETURNS TABLE(xid bigint,virtualxid int,timestampTz timestampTz,record_database text, record_user text,record_tablespace text,record_schema text, op_type text,op_text text,op_undo text,lsn pg_lsn,commit_end_lsn pg_lsn) AS 
$BODY$
DECLARE
	wal_dir_tmp  text;
	starttime_text  text;
	endtime_text  text;
BEGIN
	perform walminer_stop();
	
	if wal_dir is NULL then 
		select setting||'/pg_wal' into wal_dir_tmp from pg_settings where name='data_directory';
	else
		wal_dir_tmp := wal_dir;
	end if;
	
	perform walminer_wal_add(wal_dir_tmp);

	if startxid is NULL then
		startxid := 0;
	end if;

	if endxid is NULL then
		endxid := 0;
	end if;

	perform walminer_start('NULL', 'NULL', startxid, endxid, tempresult, use_history_image);
	return query select * from walminer_contents order by commit_end_lsn,lsn;
	
	perform walminer_stop();
END;
$BODY$
LANGUAGE 'plpgsql' VOLATILE;

